feat(cudf): Stream full-partition COUNT and peer-aware RANGE SUM - #2

Summary

• Stream full-partition constant COUNT over sorted partition input.
• Stream peer-aware multi-key RANGE UNBOUNDED PRECEDING ... CURRENT ROW running SUM.
• Bound the active partition or peer with the existing Parquet spill/replay lifecycle.
• Allow spill-safe MAP columns as OrderBy payload while continuing to reject MAP sort keys.

Companion planner PR: NVIDIA/spark-gluten#50.

Root cause

The cuDF Window adapter supported streaming rank functions but did not have bounded cross-batch state for a result that is known only at a partition boundary, or for a RANGE result that is known only at a peer boundary. Full-partition COUNT was therefore rejected, and the multi-key default Spark RANGE frame was unsupported.

The Job 144 input also carries a MAP payload through its required OrderBy. The spill schema check rejected MAP columns even when they were not sort keys.

Design

Full-partition COUNT

The operator emits completed partitions as soon as the next partition boundary arrives. It retains only the active partition rows and a BIGINT row count. Active rows spill through the existing Parquet infrastructure above the configured memory bound, then replay through the normal getOutput() backpressure path with the final count attached.

Peer-aware RANGE SUM

The operator retains the active peer, the current partition's cumulative sum, and a valid-value count. It emits a completed peer when the next peer or partition arrives. The valid count preserves Spark null behavior for all-null prefixes.

The q3 libcudf dependency does not expose the newer multi-order-column grouped RANGE API. The implementation therefore uses libcudf sorted grouping and lower-bound primitives over all partition and order columns, including direction and null ordering, and repeats each peer-end running result across the peer. It does not implement host-side RANGE comparison semantics.

All persistent GPU state is allocated on the existing state stream with the configured async RMM resource. The implementation does not add MPP control state or query rewrites.

Supported scope

• COUNT(non-null constant) with a full-partition ROWS frame and BIGINT result.
• SUM(field) with RANGE UNBOUNDED PRECEDING ... CURRENT ROW.
• Multiple order keys with ASC/DESC and NULLS FIRST/LAST.
• TINYINT, SMALLINT, INTEGER, BIGINT, and DOUBLE SUM inputs.

The adapter continues to reject nullable-column counts, DISTINCT, bounded RANGE offsets, unvalidated decimal/ANSI overflow behavior, and incompatible mixed frames.

Validation

• Final adapter test binary: 16/16 passed.
  • Cross-batch full-partition COUNT.
  • Output before noMoreInput.
  • Forced active-partition spill and cleanup.
  • Multi-key RANGE peers, ASC/DESC, null keys, null values, and all-null prefixes.
  • Cross-batch and forced active-peer spill.
  • Five existing streaming rank regressions.
• OrderBy MAP payload tests: 2/2 passed.
• Spark 4 / Scala 2.13 Maven packaging passed with the companion planner PR.
• Full Job 144 validation used this feature stack with lifecycle patch fix(cudf): Restore Spark workload coverage #54 applied locally; fix(cudf): Restore Spark workload coverage #54 is intentionally not included in this PR.
• Under that validation stack, Job 144 completed twice under strict four-endpoint MPP with native cuDF, async allocation, and CPU fallback disabled.
  • Run 1: 51,528,228 rows in 749.259 seconds.
  • Run 2: 51,528,228 rows in 762.836 seconds.
  • Both runs reported zero fallback hits and cleaned spill state.
  • Physical plans contained native CudfOrderBy -> CudfWindow pipelines for the original full-partition COUNT and default multi-key RANGE SUM.

Additional observations

The full Job 144 query contains a later ROW_NUMBER deduplication whose order key is not unique. Order-independent output digests differ between repeated runs, and the same affected columns already differed in pre-feature runs that used the historical COUNT rewrite and explicit ROWS frame. Exact Window semantics are therefore asserted by the targeted reference tests; the full job validates native planning, row count, strict execution, and lifecycle.

Two current q3 Job 8 reruns reached an existing global GPU-memory cliff while several rank pipelines fed concurrent hash-join build states. The device reached approximately 33.1 GiB before the next OrderBy allocation failed. The new COUNT and RANGE modes are both inactive for that rank-only plan, and the existing rank behavior is unchanged by this patch.